package org.jdcenter.rmi.impl;

import java.rmi.RemoteException;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

import static org.jdcenter.rmi.impl.Glossary.*;

import net.sf.cglib.proxy.Enhancer;

import org.jdcenter.rmi.JDSCallback;
import org.jdcenter.rmi.abs.*;


public class LocalChannelFactory implements ChannelFactory {
    private static Logger aLogger = Logger.getLogger(LocalChannelFactory.class.getCanonicalName());
    private Logger getLogger() {
        return aLogger;
    }

    private Scheme context;
    private ScheduledExecutorService executor;
    private ConcurrentMap<String, RemoteObject> remotes;
    private ConcurrentMap<String, Queue<LocalObject>> channels;

    private class RemoteObject implements Runnable {
        private RemoteChannel channel;
        private AtomicInteger counter;
        private ScheduledFuture future;

        public RemoteObject(RemoteChannel channel) {
            this.channel = channel;
            this.counter = new AtomicInteger(1);

            int period = context.getProperty(CHANNEL_TICKER_PERIODS);
            int initial = context.getProperty(CHANNEL_TICKER_INITIAL_DELAY);

            this.future = executor.scheduleWithFixedDelay(this, initial, period, TimeUnit.SECONDS);
        }

        public void increment() {
            this.counter.getAndIncrement();
        }

        public void decrement() {
            this.counter.getAndDecrement();
        }

        public RemoteChannel getChannel() {
            return this.channel;
        }

        public void run() {
            try {
                int now = this.counter.get();
                if(now > 0) {
                    try {
                        this.channel.tick();
                        this.counter.compareAndSet(1, 0);
                    } catch(RemoteException e) {
                        getLogger().warning(Thrower.toString(e));
                    }
                } else {
                    remotes.remove(channel.getUrl());
                    this.future.cancel(false);
                    this.channel.close();
                }
            } catch (Exception e) {
                getLogger().severe("Oh, unexpected exception occured !!!");
                getLogger().severe(Thrower.toString(e));
            }
        }
    }

    private interface Freeable {
        void free() throws ChannelException;
    }

    private class PeerObject implements Channel, Freeable {
        private PeerChannel channel;
        private RemoteObject remote;

        public PeerObject(RemoteObject remote) {
            this.remote = remote;
            this.remote.increment();
            this.channel = new PeerChannel(remote.getChannel());
        }

        public Object invoke(String method,Object[] args,Object result) throws ChannelException, HandlerException {
            if (this.channel == null)
                throw new ChannelException("PeerChannel is closed");
            return this.channel.invoke(method,args,null);
        }

        public void free() {
            this.channel = null;
            this.remote.decrement();
        }
    }

    class LocalObject implements Runnable, Channel, Freeable {
        private int reserved;
        private LocalChannel channel;
        private RemoteObject remote;
        private Object obj;

        public LocalObject(RemoteObject remote, JDSCallback callback) throws RemoteException {
            assert(callback != null);
            this.remote = remote;
            this.obj = ((Result)remote.getChannel().lookup()).getResultObj();
            this.remote.increment();
            this.channel = new LocalChannel(remote.getChannel(), callback);
            this.reserved = context.getProperty(LOCAL_CHANNEL_RESERVED_AFTER_FREE);
        }

        private synchronized LocalChannel getChannel() throws ChannelException {
            if(this.channel == null)
                throw new ChannelException("LocalChannel is closed");

            return this.channel;
        }

        private synchronized LocalChannel removeChannel() {
            assert this.channel != null;
            if(this.channel.getCallback() != null)
                return null;

            LocalChannel x = this.channel;
            this.channel = null;
            return x;
        }

        public void reuse(JDSCallback callback) throws ChannelException {
            assert callback != null;
            this.getChannel().setCallback(callback);
        }

        public Object invoke(String method,Object[] args,Object result) throws ChannelException, HandlerException {
            return this.getChannel().invoke(method,args,null);
        }

        public void free() throws ChannelException {
            this.getChannel().setCallback(null);
            LocalChannelFactory.this.offer(this);
            executor.schedule(this, this.reserved, TimeUnit.SECONDS);
        }

        public String getUrl() {
            return this.remote.getChannel().getUrl();
        }
        
        public Object getObj(){
        	return this.obj;
        }

        public void run() {
            LocalChannel channel = this.removeChannel();
            if(channel != null) {
                LocalChannelFactory.this.cancel(this);
                channel.close();
                this.remote.decrement();
            }
        }
    }

    public LocalChannelFactory(Scheme context) {
        this.context = context;

        int pool = this.context.getProperty(CLIENT_HANDLER_THREAD_COUNT);
        this.executor = Executors.newScheduledThreadPool(pool);

        this.remotes = new ConcurrentHashMap<String, RemoteObject>();
        this.channels = new ConcurrentHashMap<String, Queue<LocalObject>>();
    }

    private void offer(LocalObject channel) {
        String url = channel.getUrl();
        Queue<LocalObject> channels = this.channels.get(url);
        assert channels != null;
        channels.offer(channel);
    }

    private void cancel(LocalObject channel) {
        String url = channel.getUrl();
        Queue<LocalObject> channels = this.channels.get(url);
        assert channels != null;
        channels.remove(channel);
    }

    private Object alloc_i(String url, JDSCallback callback) throws URLException, ChannelException {
        try {
            RemoteObject remote = this.remotes.get(url);
            if (remote == null) {
                remote = new RemoteObject(new RemoteChannel(url));
                this.remotes.putIfAbsent(url, remote);
                remote = this.remotes.get(url); // Get it again
            }

            if(callback == null)
                return new PeerObject(remote);

            Queue<LocalObject> channels = this.channels.get(url);
            if (channels == null) {
                channels = new ConcurrentLinkedQueue<LocalObject>();
                this.channels.putIfAbsent(url, channels);
                channels = this.channels.get(url); // Get it again
            }

            LocalObject channel = channels.poll();
            if (channel == null)
                channel = new LocalObject(remote, callback);
            else
                channel.reuse(callback);

            RMIInterceptor interceptor = new RMIInterceptor(channel);
			
            Enhancer e = new Enhancer();
            e.setSuperclass(channel.getObj().getClass());
            System.out.println(channel.getObj().getClass().getName());
            e.setCallback(interceptor);
            return e.create();

        } catch (RemoteException e) {
            throw new ChannelException(e);
        }
    }

    public Object alloc(String url) throws URLException, ChannelException {
        return alloc_i(url, null);
    }

    public Object alloc(String url, JDSCallback callback) throws URLException, ChannelException {
        return alloc_i(url, callback);
    }

    public void free(Channel channel) throws ChannelException {
        if (channel instanceof Freeable) {
            ((Freeable) channel).free();
        } else {
            String msg = "TextArgument " + channel + "is not an instance of Freeable";
            throw new IllegalArgumentException(msg);
        }
    }
}

